package hit.edu.zjc.Operator;

import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.UserInterface.UDColumInfo;
import hit.edu.zjc.UserInterface.UserDef;
import hit.edu.zjc.test.OperatorObjectRW;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Random;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.Integer64SerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
/*
 *建立每个NC的 (1、2)GramMap
 *HashTable<String,HashSet<Integer>>[] GramMap=new HashTable<String,HashSet<Integer>>[];
 *String 是Gram，HashSet<Integer>是含有该Gram的第一列的ID号。
 *为每个关键列都建立一个HashTable
 * */
public class Build2GramMapOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {

    private static final long serialVersionUID = 1L;
    private ISerializerDeserializer[] DataISD_array=null;
    private UDColumInfo[] udci_array=null;
    private String ByteFilePath=null;
    private Hashtable<String,HashSet<Integer>>[] GramMap=null;
    public Build2GramMapOperatorDescriptor(JobSpecification spec,ISerializerDeserializer[] isd_array,UDColumInfo[] udcis,String filepath) {
        super(spec, 1, 1);//输入输出端的数量。
        DataISD_array=isd_array;
        udci_array=udcis;
        ByteFilePath=filepath;
        GramMap=(Hashtable<String, HashSet<Integer>>[])Array.newInstance(Hashtable.class, udci_array.length);
        //进行初始化操作。Hashtable是线程安全的。
        for(int i=0;i<udci_array.length;i++)
        	GramMap[i]=new Hashtable<String,HashSet<Integer>>();
       /*不需要检查
        //检查udci_array是否按ColumID递增顺序排列。
        for(int i=1;i<udci_array.length;i++)
        	{
        		for(int j=i;j>0;j--)
        		{
        			if(udci_array[j].ColumID>udci_array[j-1].ColumID)
        				break;
        			else
        			{
        				UDColumInfo udci=udci_array[j];
        				udci_array[j]=udci_array[j-1];
        				udci_array[j-1]=udci;
        			}
        		}
         	}
        */
    }

    @Override
    public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {

        final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), new RecordDescriptor(DataISD_array));

        return new AbstractUnaryInputUnaryOutputOperatorNodePushable() {

            private ByteBuffer outputBuffer;

            private FrameTupleAppender outputAppender;

            @Override
            public void open() throws HyracksDataException {
                outputBuffer = ctx.allocateFrame();
                writer.open();
            }

            @Override
            public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
                accessor.reset(buffer);//buffer内装的是输入的数据
                int tupleCount = accessor.getTupleCount();
                int fieldCount = accessor.getFieldCount();
                byte[] byte_array=accessor.getBuffer().array();
                //accessor.prettyPrint();
                int fieldSlotsLength=accessor.getFieldSlotsLength();
                for (int i = 0; i < tupleCount; i++) {
                    int tupleStartOffset = accessor.getTupleStartOffset(i);//获得第i个Tuple的开始位置。
                    	 //对每个tuple中的关键field建立1.5Gram表：
                    int Data_ID=0;//数据条目的主键值。
                    
	                      //获取数据条目DataID的值：
						  int fieldstartoffset=accessor.getFieldStartOffset(i, 0);
						  int fieldendoffset=accessor.getFieldEndOffset(i, 0);
						  int fieldlength=accessor.getFieldLength(i, 0);   
						  String data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( 
								new DataInputStream(
										new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
						  Data_ID=Integer.parseInt(data_str);//记录数据条目的主键值。
                    
						  	 
	                    for(int k=0;k<udci_array.length;k++)
	                    	 {
	                    	try{	
	                    	int fIdx=udci_array[k].ColumID;
	                    	fieldstartoffset=accessor.getFieldStartOffset(i, fIdx);
	                    	fieldendoffset=accessor.getFieldEndOffset(i, fIdx);
	                    	fieldlength=accessor.getFieldLength(i, fIdx);
	                    	    //将field从二进制中提取出来。
	                    		
								 	data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( new DataInputStream(new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
	                    	   
	                    		//字符串的忽略大小写
	                    	data_str=data_str.toUpperCase();
	                    	    //去掉字符串空格。
	                    	data_str=data_str.replaceAll(" ","");
	                    		//将字符串，每个一个字符都加入GramMap中。
	                    	for(int data_str_i=0;data_str_i<data_str.length();data_str_i++)
	                    		{
	                    		String key=data_str.substring(data_str_i, data_str_i+1);
	                    		HashSet<Integer> Set=GramMap[k].get(key);
	                    		if(Set==null)//键不存在。
	                    			{
	                    			Set=new HashSet<Integer>();
	                    			}
                    			Set.add(Data_ID);
                    			GramMap[k].put(key, Set);
	                    		}
	                    	   //将字符串，每个2-gram都加入GramMap中
	                    	for(int data_str_i=0;data_str_i<data_str.length()-1;data_str_i++)
	                    		{
	                    		String key_2=data_str.substring(data_str_i, data_str_i+2);
	                    		HashSet<Integer> Set=GramMap[k].get(key_2);
	                    		if(Set==null)
	                    			{
	                    			Set=new HashSet<Integer>();
	                    			}
	                    		Set.add(Data_ID);
	                    		GramMap[k].put(key_2, Set);
	                    		}
	                    	 }
	                    catch (Exception e) {
									e.printStackTrace();//产生异常的原因是字符串编码不是UTF-8的。
							 }
	                    	}
                	}
            }

            @Override
            public void fail() throws HyracksDataException {
            }

            @Override
            public void close() throws HyracksDataException {
               
                byte[] objbytes=ObjectConvert.ObjectToByteArray(GramMap);

                
                if(objbytes.length<outputBuffer.capacity())
                	{
                    outputBuffer.put(objbytes);
                    FrameUtils.flushFrame(outputBuffer, writer);
                	}
                else//一个frame装不下
                	{
                		int byteremain=objbytes.length;
                		int i=0;
                		for(i=0;byteremain>=outputBuffer.capacity();i++)
                			{
                			outputBuffer.put(objbytes,i*outputBuffer.capacity(),outputBuffer.capacity());
                    		FrameUtils.flushFrame(outputBuffer, writer);
                    		outputBuffer.clear();
                    		byteremain=byteremain-outputBuffer.capacity();
                			}
                		if(byteremain>0)
                			{
                			outputBuffer.put(objbytes,i*outputBuffer.capacity(),byteremain);
                			FrameUtils.flushFrame(outputBuffer, writer);
                			}
                	}
                
                outputAppender = null;
                outputBuffer = null;
                writer.close();
            }
        };
    }
}